package com.yunhe.realtime.service.data.impl;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.constant.CacheConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.exception.ArgumentErrorException;
import com.yunhe.common.exception.ObjectNotFoundException;
import com.yunhe.common.exception.UniqueConstraintsException;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.realtime.service.data.RealtimeService;
import com.yunhe.common.util.DateTimeUtils;
import com.yunhe.common.util.RedisClient;
import com.yunhe.common.util.StringUtil;
import com.yunhe.common.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * 实时数据Service
 *
 * @author liuronglei
 * @desc 实时数据服务类实现
 */
@Service
public class RealtimeServiceImpl implements RealtimeService {
    private static final ObjectMapper mapper = new ObjectMapper();
    private static final Logger logger = LoggerFactory.getLogger(RealtimeServiceImpl.class);

    @Autowired
    private RedisClient redisClient;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 获得量测实时值
     * @param pnList 点号列表
     */
    @Override
    public Map<Long, Double> findMeasureRealtime(List<Long> pnList) {
        return redisClient.getMeasurementDouble(pnList);
    }

    /**
     * 获得量测实时值_点号为字符串的形式
     * @param pnList 点号列表
     */
    @Override
    public Map<String, Double> findMeasureRealtimeByStr(List<String> pnList) {
        return redisClient.getMeasurementDoubleByStr(pnList);
    }

    /**
     * 获得量测实时对象
     * @param pnList 点号列表
     */
    @Override
    public Map<Long, Measurement> findMeasureRealtimeObj(List<Long> pnList) {
        return redisClient.getMeasurementObject(pnList);
    }

    /**
     * 获得量测实时对象_字符串形式
     * @param pnList 点号列表
     */
    @Override
    public Map<String, Measurement> findMeasureRealtimeObjByStr(List<String> pnList) {
        return redisClient.getMeasurementObjectByStr(pnList);
    }

    /**
     * 保存量测实时值
     *
     * @param pointNumber 点号
     * @param value       量测值
     */
    @Override
    public void saveMeasureRealtime(Long pointNumber, Double value) {
        //去除直接保存值的方式，只保留保存对象类型的方式
//        redisClient.set(String.valueOf(pointNumber), String.valueOf(value));
        //由于上海电站站端目前没有实现保存带时间的实时量测数据，所以由云端在保存量测实时值时保存一份带时间戳的量测对象实时值
        saveMeasureRealtimeObj(pointNumber, value);
    }

    /**
     * 批量保存量测实时值
     *
     * @param pointNumberValueMap 点号和值对应Map
     */
    @Override
    public void saveMeasureRealtime(Map<Long, Double> pointNumberValueMap)
            throws ArgumentErrorException, UniqueConstraintsException, ObjectNotFoundException {
        if (pointNumberValueMap.size() == 1) {
            for (Map.Entry<Long, Double> entry : pointNumberValueMap.entrySet()) {
                redisClient.set(String.valueOf(entry.getKey()), String.valueOf(entry.getValue()));
            }
        } else {
            redisClient.setMeasurementDouble(pointNumberValueMap);
        }

//        for (Map.Entry<Long, Double> entry : pointNumberValueMap.entrySet()) {
//            redisClient.set(String.valueOf(entry.getKey()), String.valueOf(entry.getValue()));
//        }

        /*
        Long[] pointNumbers = dataMonitorConfig.getFrequency();
        Map<Long, Double> map = new HashMap<>();
        for (Long pn : pointNumbers) {
            map.put(pn, 0D);
        }
        List<Measurement> measurements = new ArrayList<>();
        for (Map.Entry<Long, Double> entry : pointNumberValueMap.entrySet()) {
            redisClient.set(String.valueOf(entry.getKey()), String.valueOf(entry.getValue()));
            if (map.containsKey(entry.getKey())) {
                Measurement measurement = new Measurement();
                measurement.setVal(entry.getValue());
                measurement.setPointNumber(entry.getKey());
                measurement.setDtime(new Date());
                measurements.add(measurement);
            }
        }
        measurementService.createMeasurementAsync(measurements);
        */
    }

    /**
     * 批量保存量测实时值（异步方法）
     *
     * @param pointNumberValueMap 点号和值对应Map
     */
    @Async
    @Override
    public void saveMeasureRealtimeAsync(Map<Long, Double> pointNumberValueMap)
            throws ArgumentErrorException, UniqueConstraintsException, ObjectNotFoundException {
        //由于上海电站站端目前没有实现保存带时间的实时量测数据，所以由云端在保存量测实时值时保存一份带时间戳的量测对象实时值
        saveMeasureRealtimeObj(pointNumberValueMap);
        //todo 由于上海电站站端程序还未发布，暂时调用Kafka保存历史量测数据，站端程序发布以后要去掉
        if (pointNumberValueMap != null) {
            Set<Long> keys = pointNumberValueMap.keySet();
            if (keys.size() > 0) {
                Long pointNumber = keys.iterator().next();
                String pidStr = String.valueOf(pointNumber);
                if (pidStr.startsWith("1000") || pidStr.startsWith("1102")) {
                    try {
                        String data = mapper.writeValueAsString(pointNumberValueMap);
                        kafkaTemplate.send(KafkaConsts.Topic.Realtime + "Shanghai", data);
                    } catch (JsonProcessingException e) {
                        logger.error("量测更新实时推送报错：{}", pointNumberValueMap.toString(), e);
                    }
                }
            }
        }
    }

    private void saveMeasureRealtimeObj(Long pointNumber, Double value) {
        if (pointNumber == null) {
            return;
        }
        try {
            long rtime = System.currentTimeMillis();
            String timeZoneId = null;
            String pid = String.valueOf(pointNumber);
            if (pid != null && pid.startsWith("1108")) {
                timeZoneId = "Australia/Queensland";
            }
            String dtime = DateTimeUtils.getDateString(rtime, timeZoneId);
            Measurement measurement = new Measurement();
            measurement.setPointNumber(pointNumber);
            measurement.setVal(value);
            measurement.setDtime(dtime);
            measurement.setRtime(rtime);
            saveMeasureRealtimeObj(Collections.singletonList(measurement));
        } catch (Exception e) {
            logger.error("保存量测实时值对象报错", e);
        }
    }

    private void saveMeasureRealtimeObj(Map<Long, Double> pointNumberValueMap) {
        if (pointNumberValueMap == null || pointNumberValueMap.size() == 0) {
            return;
        }
        long rtime = System.currentTimeMillis();
        List<Measurement> list = new ArrayList<>();
        try {
            String timeZoneId = null;
            for (Map.Entry<Long, Double> entry : pointNumberValueMap.entrySet()) {
                Long key = entry.getKey();
                String pid = String.valueOf(key);
                if (pid != null && pid.startsWith("1108")) {
                    timeZoneId = "Australia/Queensland";
                }
                break;
            }
            String dtime = DateTimeUtils.getDateString(rtime, timeZoneId);
            pointNumberValueMap.forEach((k, v) -> {
                Measurement measurement = new Measurement();
                measurement.setPointNumber(k);
                measurement.setVal(v);
                measurement.setDtime(dtime);
                measurement.setRtime(rtime);
                list.add(measurement);
            });
            saveMeasureRealtimeObj(list);
        } catch (Exception e) {
            logger.error("保存量测实时值对象报错", e);
        }
    }

    /**
     * 保存量测实时对象
     *
     * @param measurement 量测对象
     */
    @Override
    public void saveMeasureRealtimeObj(Measurement measurement) {
        saveMeasureRealtimeObj(Collections.singletonList(measurement));
    }

    /**
     * 批量保存量测实时对象
     *
     * @param list 量测对象列表
     */
    @Override
    public void saveMeasureRealtimeObj(List<Measurement> list) {
        if (CollectionUtil.isNotEmpty(list)) {
            if (list.size() == 1) {
                redisClient.setMeasurementObject(list.get(0));
            } else {
                redisClient.setMeasurementObject(list);
            }
            try {
                kafkaTemplate.send(KafkaConsts.Topic.Realtime, mapper.writeValueAsString(list));
            } catch (JsonProcessingException e) {
                logger.error("量测更新实时推送报错：{}", list.toString());
                e.printStackTrace();
            }
        }
    }

    /**
     * 批量保存量测实时对象（异步方法）
     *
     * @param list 量测对象列表
     */
    @Async
    @Override
    public void saveMeasureRealtimeObjAsync(List<Measurement> list) {
        saveMeasureRealtimeObj(list);
    }

    /**
     * 删除量测实时对象（异步方法）
     * @param pointNumbers 点号列表
     */
    @Async
    @Override
    public void delAsync(List<Long> pointNumbers) {
        List<String> keys = new ArrayList<>();
        for (Long pointNumber : pointNumbers) {
            keys.add(CacheConsts.KEY.MEASURE_OBJ.value(pointNumber));
        }
        redisClient.del(keys);
    }

    /**
     * 删除量测实时对象的备份（异步方法）
     * @param pointNumbers 点号列表
     */
    @Async
    @Override
    public void delBackupAsync(List<Long> pointNumbers) {
        List<String> keys = new ArrayList<>();
        for (Long pointNumber : pointNumbers) {
            keys.add(CacheConsts.KEY.MEASURE_BACKUP.value(pointNumber));
        }
        redisClient.del(keys);
    }

    /**
     * 将量测实时对象转为备份（异步方法）
     * @param pointNumbers 点号列表
     */
    @Async
    @Override
    public void movetoBackupAsync(List<Long> pointNumbers) {
        List<String> keys = new ArrayList<>();
        List<String> newKeys = new ArrayList<>();
        for (Long pointNumber : pointNumbers) {
            keys.add(CacheConsts.KEY.MEASURE_OBJ.value(pointNumber));
            newKeys.add(CacheConsts.KEY.MEASURE_BACKUP.value(pointNumber));
        }
        redisClient.rename(keys, newKeys);
    }

    /**
     * 获得量测实时对象备份
     * @param pointNumbers 点号列表
     */
    @Override
    public Map<Long, Double> getBackup(List<Long> pointNumbers) {
        if (pointNumbers == null) {
            return null;
        }
        List<String> keys = new ArrayList<>();
        pointNumbers.forEach(v -> keys.add(CacheConsts.KEY.MEASURE_BACKUP.value(v)));
        List<String> values = redisClient.mget(keys);
        Map<Long, Double> result = new HashMap<>(pointNumbers.size());
        for (int i = 0; i < pointNumbers.size(); i++) {
            String value = values.get(i);
            if (!StringUtil.isEmpty(value)) {
                Measurement measurement = JSONObject.parseObject(value, Measurement.class);
                if (measurement != null) {
                    result.put(pointNumbers.get(i), measurement.getVal());
                }
                else {
                    result.put(pointNumbers.get(i), null);
                }
            } else {
                result.put(pointNumbers.get(i), null);
            }
        }
        return result;
    }

    /**
     * 更新量测实时对象
     * @param pointNumber 点号
     * @param rtime 时间戳
     * @param dtime 时间字符串
     * @param val 值
     * @param flag 标记
     */
    @Override
    public void updateMeasureRealtime(Long pointNumber, Long rtime, String dtime, Double val, Integer flag) {
        Measurement measurement = redisClient.getMeasurementObj(pointNumber);
        if (measurement == null) {
            measurement = new Measurement();
            measurement.setPointNumber(pointNumber);
        }
        if (rtime != null) {
            measurement.setRtime(rtime);
        }
        if (!StringUtil.isEmpty(dtime)) {
            measurement.setDtime(dtime);
        }
        if (val != null) {
            measurement.setVal(val);
        }
        if (flag != null) {
            measurement.setFlag(flag);
        }
        redisClient.setMeasurementObject(measurement);
    }

    /**
     * 添加值
     *
     * @param key      键
     * @param value    值
     * @param liveTime 过期时间（秒）
     */
    @Override
    public void setString(String key, String value, Optional<Integer> liveTime) {
        if (liveTime.isPresent()) {
            redisClient.set(key, String.valueOf(value), liveTime.get());
        } else {
            redisClient.set(key, String.valueOf(value));
        }
    }

    /**
     * 获得值
     *
     * @param key 键
     */
    @Override
    public String getString(String key) {
        return redisClient.get(key);
    }

    /**
     * 添加对象
     *
     * @param key      键
     * @param value    值
     * @param liveTime 过期时间（秒）
     */
    @Override
    public void setMap(String key, Map value, Optional<Integer> liveTime) {
        if (liveTime.isPresent()) {
            redisClient.setObject(key, value, liveTime.get());
        } else {
            redisClient.setObject(key, value);
        }
    }

    /**
     * 获得对象
     *
     * @param key 键
     */
    @Override
    public Map getMap(String key) {
        return redisClient.getObject(key, Map.class);
    }

    /**
     * 删除值
     *
     * @param key 键
     */
    @Override
    public void delString(String key) {
        redisClient.del(key);
    }

    /**
     * 是否存在键
     *
     * @param key 键
     */
    @Override
    public boolean hasKey(String key) {
        return redisClient.exists(key);
    }

}
